package online.shenjian.mq;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;
import org.springframework.amqp.rabbit.connection.Connection;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactoryUtils;
import org.springframework.amqp.rabbit.core.RabbitMessagingTemplate;
import org.springframework.amqp.rabbit.core.RabbitOperations;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;

@Component
@ConditionalOnProperty(name = "spring.rabbitmq.enable", matchIfMissing = true)
public class PublishConfirmSender {

    private RabbitTemplate simpleRabbitTemplate;

    private RabbitTemplate asyncRabbitTemplate;

    private ConnectionFactory myRabbitConnectionFactory;

    public PublishConfirmSender(RabbitTemplate simpleRabbitTemplate, RabbitTemplate asyncRabbitTemplate, ConnectionFactory myRabbitConnectionFactory) {
        this.simpleRabbitTemplate = simpleRabbitTemplate;
        this.asyncRabbitTemplate = asyncRabbitTemplate;
        this.myRabbitConnectionFactory = myRabbitConnectionFactory;
    }

    public void oneSender() {
        boolean sendFlag = simpleRabbitTemplate.invoke(operations -> {
            simpleRabbitTemplate.convertAndSend("direct", "orange", "orange msg");
            return simpleRabbitTemplate.waitForConfirms(5000);
        });
        if (sendFlag) {
            System.out.println("消息已成功发送");
        }
    }

    public void batchSender() {
        boolean sendFlag = simpleRabbitTemplate.invoke(operations -> {
            for (int i = 0; i < 50; i++) {
                simpleRabbitTemplate.convertAndSend("direct", "orange", "orange " + i + "msg");
                if (i % 10 == 0) {
                    if (simpleRabbitTemplate.waitForConfirms(5000)) {
                        System.out.println(i / 10 + "批次消息已全部成功发送");
                    }
                }
            }
            return simpleRabbitTemplate.waitForConfirms(5000);
        });
        if (sendFlag) {
            System.out.println("消息已全部成功发送");
        }
    }

    public void asyncSender() {
        asyncRabbitTemplate.invoke(operations -> {
            for (int i = 0; i < 50; i++) {
                String body = "orange " + i + "msg";
                simpleRabbitTemplate.convertAndSend("direct", "orange", body);
            }
            return null;
        }, (deliveryTag, multiple) -> {
            System.out.format("消息已确认. Sequence number: %d, multiple: %b%n", deliveryTag, multiple);
        }, (deliveryTag, multiple) -> {
            System.err.format("消息未确认. Sequence number: %d, multiple: %b%n",deliveryTag, multiple);
        });
    }
}
